package com.allen.flink.batch.FaultTolerace;

import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.restartstrategy.RestartStrategies;
import org.apache.flink.api.common.time.Time;
import org.apache.flink.api.java.DataSet;
import org.apache.flink.api.java.ExecutionEnvironment;

import java.util.concurrent.TimeUnit;

/**
 * 功能:批处理容错
 *
 * @date: 2020-03-24 15:23
 * @author: Allen
 * @version: 0.0.4-snapshot
 * @Email: allenZyhang@163.com
 * @since: JDK 1.8
 **/
public class FaultTolerance {
    
    public static void main(String[] args) throws Exception {
        //获取执行环境
        final ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
        
        //设置容错
        
        // 固定延时   重试 3次  每隔10s
        // env.setRestartStrategy(RestartStrategies.fixedDelayRestart(3, Time.of(10, TimeUnit.SECONDS)));
        
        //失败率  1小时内 只能失败2次  执行间隔10s
        env.setRestartStrategy(RestartStrategies.failureRateRestart(2, Time.of(1, TimeUnit.HOURS), Time.of(10, TimeUnit.SECONDS)));
        
        //读取数据
        DataSet<String> data = env.fromElements("1", "2", "", "4", "5");
        
        data.map(new MapFunction<String, Integer>() {
            @Override
            public Integer map(final String value) throws Exception {
                return Integer.parseInt(value);
            }
        }).print();
    }
    
}
